package org.boot.mq;

import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.List;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

public class MqConsumer {
	
	public static void main(String[] args) throws MQClientException {
		
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ProducerGroup");
		consumer.setNamesrvAddr("10.0.14.16:9876");
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		consumer.subscribe("TopicTest","*");
		consumer.registerMessageListener(new MessageListenerConcurrently(){
		    
		    @Override
		    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context){
		        
		        for(MessageExt ext:msgs){
		            try {
		            	Thread.sleep(1000);
						System.out.println(new Date()+"-------"+ new String(ext.getBody(),"utf-8"));
					} catch (UnsupportedEncodingException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
		        }
		        
		        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		        
		    }
		    
		    
		    
		});

		consumer.start();

		System.out.println("Consumer start ....");
		
		
		
		
	}
}	
